package cn.hy.dataSync.job;


import cn.hy.dataSync.config.*;
import cn.hy.dataSync.constant.Delimiter;
import cn.hy.dataSync.job.scheduler.AbstractScheduler;
import cn.hy.dataSync.job.scheduler.JobScheduler;
import cn.hy.dataSync.statistics.VMInfo;
import cn.hy.dataSync.statistics.communication.Communication;
import cn.hy.dataSync.statistics.communication.CommunicationTool;
import cn.hy.dataSync.statistics.container.communicator.AbstractContainerCommunicator;
import cn.hy.dataSync.statistics.container.communicator.JobContainerCommunicator;
import cn.hy.dataSync.task.ErrorRecordChecker;
import cn.hy.dataSync.utils.RdbUtils;
import com.google.common.base.Joiner;
import com.google.common.collect.Lists;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.Validate;
import org.springframework.util.CollectionUtils;

import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;

/**
 * job 容器，
 */
@Slf4j
public class JobContainer extends AbstractContainer {
    private static final int DEFAULT_CHANNEL = 2;
    private static final long DEFAULT_ERROR_RECORD = 1L;

    private static final SimpleDateFormat DATE_FORMAT =
            new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

    private long startTimeStamp;

    private long endTimeStamp;

    private long startTransferTimeStamp;

    private long endTransferTimeStamp;

    private int totalStage = 1;

    private List<TaskConfig> tasks;

    private ErrorRecordChecker errorLimit;

    private final Thread shutdownHook = new Thread(() -> {
        //死于襁褓
        if (CollectionUtils.isEmpty(this.tasks) || Objects.isNull(super.getContainerCommunicator())) {
            return;
        }

        //任务已经完成
        if (super.getContainerCommunicator().collect().isFinished()) {
            return;
        }

        endTimeStamp = System.currentTimeMillis();

        log.info("迁移任务被中断，快照当前状态：");

        //出师未捷身先死，尚留功名后人知
        for (TaskConfig taskConfig : this.tasks) {
            log.info("任务ID：{}\n\t数据读取SQL：{}\n\t状态快照：{}",
                    taskConfig.getTaskId(),
                    taskConfig.getReaderSql(),
                    CommunicationTool.Jsonify.getSnapshot(
                            super.getContainerCommunicator().getCommunication(taskConfig.getTaskId())
                    )
            );
        }

        logStatistics();

        //AlertManager.sendText(String.format("迁移任务【%s】被中断，请及时处理", jobConfig.getJobId()));
    });


    public JobContainer(final JobConfig jobConfig) {
        super(jobConfig);
        this.errorLimit = new ErrorRecordChecker(jobConfig.getErrorLimit());

        Runtime.getRuntime().addShutdownHook(shutdownHook);
    }

    @Override
    public void start() {
        log.info("开始迁移任务，任务ID：[{}]", jobConfig.getJobId());
        boolean hasException = false;
        try {
            this.startTimeStamp = System.currentTimeMillis();

            log.debug("开始前置处理 ...");
            this.preHandle();

            log.debug("开始**初始化 ...");
            this.init(); // 空实现

            log.info("开始任务拆分 ...");
            this.totalStage = this.split();

            log.info("开始调度执行 ...");
            this.schedule();

            log.debug("开始后置通知 ...");
            this.post(); // 空实现

            log.debug("开始后置处理 ...");
            this.postHandle(); //空实现

            log.info("迁移任务成功结束，任务ID：[{}]", jobConfig.getJobId());
        } catch (Throwable e) {
            log.error(String.format("迁移任务执行异常，任务ID：%s", jobConfig.getJobId()), e);
            hasException = true;

            if (e instanceof OutOfMemoryError) {
                this.destroy();
                System.gc();
            }

            // 出现异常，直接创建job 容器的统计信息
            if (Objects.isNull(super.getContainerCommunicator())) {
                AbstractContainerCommunicator tempContainerCollector;
                tempContainerCollector = new JobContainerCommunicator(jobConfig);

                super.setContainerCommunicator(tempContainerCollector);
            }

            Communication communication = super.getContainerCommunicator().collect();
            communication.setThrowable(e);
            communication.setTimestamp(this.endTimeStamp);
            Communication tempComm = new Communication();
            tempComm.setTimestamp(this.startTransferTimeStamp); // 这是那里初始化？
            Communication reportCommunication =
                    CommunicationTool.getReportCommunication(communication, tempComm, this.totalStage);
            super.getContainerCommunicator().report(reportCommunication);

            throw new RuntimeException(e);
        } finally {
            this.destroy(); // 空实现
            this.endTimeStamp = System.currentTimeMillis();

            if (!hasException) {
                VMInfo vmInfo = VMInfo.getVmInfo();
                if (vmInfo != null) {
                    vmInfo.getDelta(false);
                    log.info(vmInfo.totalString());
                }
                logStatistics();
            }
            Runtime.getRuntime().removeShutdownHook(shutdownHook);
        }
    }

    /**
     * 预处理
     */
    private void preHandle() {
        preReaderHandle();
        preWriterHandle();
        // 核心配置，内部使用
        if (Objects.isNull(jobConfig.getCoreConfig())) {
            jobConfig.setCoreConfig(CoreConfig.create());
        }
        // 是否配置错误率
        if (Objects.isNull(jobConfig.getErrorLimit())) {
            jobConfig.setErrorLimit(ErrorLimit.builder()
                    .record(DEFAULT_ERROR_RECORD).build());
        }
        // 是否配置 速率
        if (Objects.isNull(jobConfig.getSpeedConfig())) {
            jobConfig.setSpeedConfig(SpeedConfig.builder()
                    .channel(DEFAULT_CHANNEL).build());
        }
    }

    /**
     * 1.空判断，空字段判断
     * 2.校验配置的属性列是否在数据库列中
     * 3.优化 where 条件和 jdbcUrl 参数优化
     */
    private void preReaderHandle() {

        ReaderConfig readerConfig = Validate.notNull(jobConfig.getReaderConfig(), "job readerConfig 不能为空");
        Validate.notNull(readerConfig.getJdbcUrl(), "job readerConfig jdbcUrl 不能为空");
        Validate.notNull(readerConfig.getUsername(), "job readerConfig username 不能为空");
        Validate.notNull(readerConfig.getPassword(), "job readerConfig password 不能为空");
        Validate.notNull(readerConfig.getTable(), "job readerConfig table 不能为空");
        Validate.notNull(readerConfig.getPkName(), "job readerConfig pkName 不能为空");
        if (CollectionUtils.isEmpty(readerConfig.getColumns())) {
            //不配置 readerConfig 列信息，默认获取所有列
            readerConfig.setColumns(Lists.newArrayList(Delimiter.ASTERISK));
        } else {
            boolean existPkColumn = false;
            for (String column : readerConfig.getColumns()) {
                Validate.notBlank(column, "job readerConfig columns 存在异常列，请检查");
                if (column.equals(readerConfig.getPkName())) {
                    existPkColumn = true;
                }
            }
            if (!existPkColumn) {
                readerConfig.getColumns().add(readerConfig.getPkName());
            } else if (readerConfig.getColumns().size() == 1) {
                //列信息就只配置了主键，等于没配
                readerConfig.setColumns(Lists.newArrayList(Delimiter.ASTERISK));
            }
        }

        if (readerConfig.getColumns().size() > 1) {
            List<String> allColumns = RdbUtils.getTableColumns(
                    readerConfig.getJdbcUrl(),
                    readerConfig.getUsername(),
                    readerConfig.getPassword(),
                    readerConfig.getTable()
            );

            for (String column : readerConfig.getColumns()) {
                Validate.isTrue(allColumns.contains(column),
                        String.format("job readerConfig 在表 %s 中不存在 %s 列，请检查",
                                readerConfig.getTable(), column));
            }
        }

        String where = readerConfig.getWhere();
        if (StringUtils.isNotBlank(where)) {
            String whereImprove = where.trim();
            if (whereImprove.endsWith(Delimiter.SEMICOLON)) {
                whereImprove = whereImprove.substring(0, whereImprove.length() - 1);
            }
            readerConfig.setWhere(whereImprove);
        }
        readerConfig.setJdbcUrl(DataBaseType.parseDataBaseTypeFromJdbcUrl(readerConfig.getJdbcUrl())
                .appendJdbcSuffix(readerConfig.getJdbcUrl()));
    }

    /**
     * 1.空判断，空字段判断
     * 2.校验配置的类是否在数据库列中，
     * 3.主键不能转换
     * 4.jdbcUrl 优化参数
     */
    private void preWriterHandle() {
        WriterConfig writerConfig = Validate.notNull(jobConfig.getWriterConfig(), "job writerConfig 不能为空");
        Validate.notNull(writerConfig.getJdbcUrl(), "job writerConfig jdbcUrl 不能为空");
        Validate.notNull(writerConfig.getUsername(), "job writerConfig username 不能为空");
        Validate.notNull(writerConfig.getPassword(), "job writerConfig password 不能为空");
        Validate.notNull(writerConfig.getTable(), "job writerConfig table 不能为空");
        Validate.notNull(writerConfig.getPkName(), "job writerConfig pkName 不能为空");
        Validate.notEmpty(writerConfig.getTransformers(), "job writerConfig transformers 不能为空");
        Validate.isTrue(1 <= writerConfig.getBatchSize(),
                String.format("您的batchSize配置有误. 您所配置的写入数据库表的 batchSize:%d 不能小于1. 推荐配置范围为：[100-1000], 该值越大, 内存溢出可能性越大. 请检查您的配置并作出修改.", writerConfig.getBatchSize()));

        List<String> allColumns = RdbUtils.getTableColumns(
                writerConfig.getJdbcUrl(),
                writerConfig.getUsername(),
                writerConfig.getPassword(),
                writerConfig.getTable()
        );

        boolean existPkTransformer = false;
        for (Transformer transformer : writerConfig.getTransformers()) {
            Validate.notBlank(transformer.getColumn(), "job writerConfig transformers 存在异常列，请检查");
            Validate.notBlank(transformer.getExpression(),
                    String.format("job writerConfig transformer [%s] 缺少转换规则", transformer.getColumn()));
            Validate.isTrue(allColumns.contains(transformer.getColumn()),
                    String.format("job writerConfig 在表 %s 中不存在 %s 列，请检查",
                            writerConfig.getTable(), transformer.getColumn()));
            if (transformer.getColumn().equals(writerConfig.getPkName())) {
                existPkTransformer = true;
            }
        }
        Validate.isTrue(existPkTransformer, "job writerConfig 不存在主键转换规则");
        writerConfig.setJdbcUrl(DataBaseType.parseDataBaseTypeFromJdbcUrl(writerConfig.getJdbcUrl())
                .appendJdbcSuffix(writerConfig.getJdbcUrl()));
    }

    private void init() {

    }

    private void postHandle() {

    }

    private int split() {
        SpeedConfig speedConfig = jobConfig.getSpeedConfig();
        ReaderConfig readerConfig = jobConfig.getReaderConfig();
        List<String> splitQuerySql = RdbUtils.splitSingleTable(
                readerConfig.getJdbcUrl(),
                readerConfig.getUsername(),
                readerConfig.getPassword(),
                readerConfig.getTable(),
                Joiner.on(Delimiter.COMMA)
                        .join(readerConfig.getColumns()),
                readerConfig.getPkName(),
                readerConfig.getWhere(),
                speedConfig.getChannel()
        );
        WriterConfig writerConfig = jobConfig.getWriterConfig();
        List<String> valueHolders = new ArrayList<>(writerConfig.getColumns().size());
        for (int i = 0; i < writerConfig.getColumns().size(); i++) {
            valueHolders.add(Delimiter.QUESTION_MARK);
        }
        String writeSql = RdbUtils.getWriteSql(
                writerConfig.getColumns(),
                valueHolders,
                writerConfig.getWriteMode().name(),
                DataBaseType.parseDataBaseTypeFromJdbcUrl(writerConfig.getJdbcUrl())
        );

        if (WriterConfig.WriteMode.UPDATE.equals(writerConfig.getWriteMode())) {
            writeSql = String.format(writeSql, writerConfig.getTable(), writerConfig.getPkName());
            for (Transformer transformer : writerConfig.getTransformers()) {
                if (transformer.getColumn().equals(writerConfig.getPkName())) {
                    writerConfig.getTransformers().add(transformer);
                    break;
                }
            }
        } else {
            // 替换表名
            writeSql = String.format(writeSql, writerConfig.getTable());
        }

        this.tasks = new ArrayList<>(splitQuerySql.size());
        for (String querySql : splitQuerySql) {
            this.tasks.add(new TaskConfig(jobConfig, querySql, writeSql));
        }

        return this.tasks.size();
    }

    private void schedule() {
        try {
            AbstractScheduler scheduler = initScheduler(jobConfig);
            this.startTransferTimeStamp = System.currentTimeMillis();
            scheduler.schedule(tasks);
            this.endTransferTimeStamp = System.currentTimeMillis();
        } catch (Exception e) {
            this.endTransferTimeStamp = System.currentTimeMillis();
            throw new RuntimeException(e);
        }

        /*
         * 检查任务执行情况
         */
        this.checkLimit();
    }

    private AbstractScheduler initScheduler(final JobConfig jobConfig) {
        super.setContainerCommunicator(new JobContainerCommunicator(jobConfig));
        return new JobScheduler(super.getContainerCommunicator());
    }

    private void post() {

    }

    private void destroy() {
    }

    private void logStatistics() {
        long totalCosts = (this.endTimeStamp - this.startTimeStamp) / 1000;
        long transferCosts = (this.endTransferTimeStamp - this.startTransferTimeStamp) / 1000;
        if (0L == transferCosts) {
            transferCosts = 1L;
        }

        if (super.getContainerCommunicator() == null) {
            return;
        }

        Communication communication = super.getContainerCommunicator().collect();
        communication.setTimestamp(this.endTimeStamp);

        Communication tempComm = new Communication();
        tempComm.setTimestamp(this.startTransferTimeStamp);

        Communication reportCommunication = CommunicationTool.getReportCommunication(communication, tempComm, this.totalStage);

        long recordSpeedPerSecond = communication.getLongCounter(CommunicationTool.READ_SUCCEED_RECORDS)
                / transferCosts;

        reportCommunication.setLongCounter(CommunicationTool.RECORD_SPEED, recordSpeedPerSecond);

        super.getContainerCommunicator().report(reportCommunication);

        log.info(String.format(
                "\n" + "%-26s: %-18s\n" + "%-26s: %-18s\n" + "%-26s: %19s\n"
                        + "%-26s: %19s\n" + "%-26s: %19s\n" + "%-26s: %19s\n",
                "任务启动时刻",
                DATE_FORMAT.format(startTimeStamp),

                "任务结束时刻",
                DATE_FORMAT.format(endTimeStamp),

                "任务总计耗时",
                String.valueOf(totalCosts) + "s",

                "记录写入速度",
                String.valueOf(recordSpeedPerSecond) + "rec/s",

                "读出记录总数",
                String.valueOf(CommunicationTool.getTotalReadRecords(communication)),

                "读写失败总数",
                String.valueOf(CommunicationTool.getTotalErrorRecords(communication))
        ));

        if (communication.getLongCounter(CommunicationTool.TRANSFORMER_SUCCEED_RECORDS) > 0
                || communication.getLongCounter(CommunicationTool.TRANSFORMER_FAILED_RECORDS) > 0) {
            log.info(String.format(
                    "\n" + "%-26s: %19s\n" + "%-26s: %19s\n" + "%-26s: %19s\n",
                    "Transformer成功记录总数",
                    communication.getLongCounter(CommunicationTool.TRANSFORMER_SUCCEED_RECORDS),

                    "Transformer失败记录总数",
                    communication.getLongCounter(CommunicationTool.TRANSFORMER_FAILED_RECORDS),

                    "Transformer过滤记录总数",
                    communication.getLongCounter(CommunicationTool.TRANSFORMER_FILTER_RECORDS)
            ));
        }
    }

    private void checkLimit() {
        Communication communication = super.getContainerCommunicator().collect();
        errorLimit.checkRecordLimit(communication);
        errorLimit.checkPercentageLimit(communication);
    }
}
